package com.diaohw.platform.utility.netty;

import com.diaohw.platform.common.util.ThreadUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class NettyServerUtils {

    /**
     * 启动netty 服务端
     */
    public static void start(int port, ChannelHandler handler) throws Exception {

        // 创建boss线程组，用于接收连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 创建worker线程组，用于处理连接上的I/O操作，含有子线程NioEventGroup个数为CPU核数大小的2倍
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        // 创建ServerBootstrap实例，服务器启动对象
        ServerBootstrap bootstrap = new ServerBootstrap();

        // 使用链式编程配置参数
        // 将boss线程组和worker线程组暂存到ServerBootstrap
        bootstrap.group(bossGroup, workerGroup);
        // 设置服务端Channel类型为NioServerSocketChannel作为通道实现
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();

                // 这里将LengthFieldBasedFrameDecoder添加到pipeline的首位，因为其需要对接收到的数据
                // 进行长度字段解码，这里也会对数据进行粘包和拆包处理
                //maxFrameLength：指定了每个包所能传递的最大数据包大小；
                //lengthFieldOffset：指定了长度字段在字节码中的偏移量；
                //lengthFieldLength：指定了长度字段所占用的字节长度；
                //lengthAdjustment：对一些不仅包含有消息头和消息体的数据进行消息头的长度的调整，这样就可以只得到消息体的数据，这里的lengthAdjustment指定的就是消息头的长度；
                //initialBytesToStrip：对于长度字段在消息头中间的情况，可以通过initialBytesToStrip忽略掉消息头以及长度字段占用的字节。
                //1024最大数据包长度,包括长度所占的字节数
                //0,因为第一字符开始就是长度
                //4消息体长度所占字节数
                //0因为第一个字符就是长度
                //4去掉长度所占字节数,获取剩下的消息体
                //注4就是下面LengthFieldPrepender中的规定长度所占的字节数
                socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                // LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
                socketChannel.pipeline().addLast(new LengthFieldPrepender(4));
                //handler一般都是放在最后面,建议一个就可以,如果业务复杂拆分多个,那在handler中调用fireXXX才会向下一个handler继续调用
                //StringDecoder要放到LengthFieldBasedFrameDecoder后面,对得到消息体的ByteBuf转码为String类型,个人认为这个比较常用
                socketChannel.pipeline().addLast("encode", new StringDecoder());
                // 添加ServerHandler到ChannelPipeline，对workerGroup的SocketChannel（客户端）设置处理器
                pipeline.addLast(handler);
            }
        });
        // 设置启动参数，初始化服务器连接队列大小。服务端处理客户端连接请求是顺序处理，一个时间内只能处理一个客户端请求
        // 当有多个客户端同时来请求时，未处理的请求先放入队列中
        bootstrap.option(ChannelOption.SO_BACKLOG, 4096);

        start(bossGroup, bootstrap, workerGroup, port);
    }


    public static Channel start(EventLoopGroup bossGroup, ServerBootstrap bootstrap, EventLoopGroup workerGroup, int port) throws Exception {

        AtomicBoolean fail = new AtomicBoolean(true);

        AtomicReference<Channel> channel = new AtomicReference<>();

        CountDownLatch launchLatch = new CountDownLatch(1);

        CompletableFuture<Void> ru = CompletableFuture.runAsync(() -> {
            // 等待服务器关闭
            try {
                ChannelFuture channelFuture = bootstrap.bind(port).sync();

                channelFuture.addListener((ChannelFutureListener) future -> {
                    fail.set(!future.isSuccess());
                    launchLatch.countDown();
                    if (future.isSuccess()) {
                        log.info("netty server started on port {}", port);

                    }
                });
                channelFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> {
                    log.info("netty server stopped");
                });

                channel.set(channelFuture.channel());
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                // 优雅地关闭boss线程组
                bossGroup.shutdownGracefully();
                // 优雅地关闭worker线程组
                workerGroup.shutdownGracefully();
            }
        });

        ThreadUtil.await(launchLatch);
        if (fail.get()) {
            ru.get();
        }

        return channel.get();
    }

}
